# -*- coding: utf-8 -*-
from airflow.utils.email import send_email
from utils.macros import cst_ds, to_cst


def email_alert(adresses):
    def _email_alert(ctx):
        """
        :param ctx: dict 包含 ds, dag_run 等所有 jinja 模板内可使用的变量
        :return:
        """
        dag_run = ctx['dag_run']
        execution_date = ctx['execution_date']
        env = ctx['var']['value'].env
        failed_tis = dag_run.get_task_instances(state=['failed', 'upstream_failed'])

        title = f'[{env.upper()}]: {dag_run.state.upper()} {dag_run.dag_id}({cst_ds(execution_date)})'
        failed_tasks_str = ("<div>   - 失败任务:<div/>" + "".join(["<div>      - " + ti.task_id + ("(上游失败)" if ti.state == 'upstream_failed' else "") + "<div/>" for ti in failed_tis])) if len(failed_tis) > 0 else ""
        text = f"""
<h1>{title}<h1/>
<div>   - <strong>状态:<strong/> {dag_run.state}<div/>
<div>   - <strong>日期:<strong/> {cst_ds(execution_date)}<div/>
<div>   - <strong>开始时间:<strong/> {to_cst(dag_run.start_date)}<div/>
<div>   - <strong>结束时间:<strong/> {to_cst(dag_run.end_date)}<div/>   
""" + failed_tasks_str

        return send_email(to=adresses,
                          subject=f'[{env}] {dag_run.state.upper()} {dag_run.dag_id}({cst_ds(execution_date)})',
                          html_content=text,
                          mime_charset='utf-8')

    return _email_alert
